package com.demo.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class MyConsumer {

    @KafkaListener(topics = "my-topic",groupId = "group1")
    public void listenGroup(ConsumerRecord<String,String> record, Acknowledgment ack) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);

        var a = "c";
        System.out.println( a instanceof String);
        // commit offset
        ack.acknowledge();
    }

    @KafkaListener(groupId = "group2",topicPartitions = {
            @TopicPartition(topic = "topic1",partitions = {"0","1"}),
            @TopicPartition(topic = "topic2",partitions = {"0"}, partitionOffsets = @PartitionOffset(partition = "1",initialOffset = "1000"))
    },concurrency = "3")
    // concurrency就是在同组下的消费组个数，并发消费数，建议小于等于分区总数
    public void listenGroup2(ConsumerRecord<String,String> record, Acknowledgment ack) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);

        // commit offset
        ack.acknowledge();
    }


    @KafkaListener(topics = "my-topic",groupId = "group2")
    public void listenGroups(ConsumerRecord<String,String> records, Acknowledgment ack) {
        //遍历records
        // commit
        ack.acknowledge();
    }

}
